#include <stdlib.h>

#include <common/macro/build.h>
#include <common/macro/array.h>

#include <query/query.h>
#include <query/filter.h>
#include <query/result.h>
#include <query/iterator.h>
#include <query/commands/delete.h>
#include <storage/storage.h>
#include <storage/page.h>
#include <storage/blob/page_blob.h>
#include <storage/tuple/page_tuple.h>
#include <transaction/transaction.h>

API_LIBLOCAL result_e query_delete_blob(query_t *query, page_number_t page_number, transaction_id_t deleter) {
	page_blob_type_e blob_type;
	page_t *page;

	do {
		page = NULL;
		storage_page_writelock(query->storage, page_number);
		storage_page_read(query->storage, page_number, &page);
		page->blob->id_delete = deleter;
		page->blob->deleted = 1;
		storage_page_write(query->storage, page_number, page);
		storage_page_rwunlock(query->storage, page_number);

		blob_type = page->blob->type;
		page_number = page->blob->next;

		page_destroy(page);

	} while (blob_type != BLOB_SINGLE && blob_type != BLOB_TAIL);

	return RESULT_OK;
}

API_LIBLOCAL result_e query_delete_iterator_openpage(query_t *query, page_number_t page_number, page_t **page) {
	storage_page_antivacuumlock(query->storage, page_number);
	storage_page_writelock(query->storage, page_number);
	storage_page_read(query->storage, page_number, page);

	query->delete_ctx.dirty = 0;
	query->delete_ctx.blockers = NULL;
	query->delete_ctx.blockers_c = 0;

	return RESULT_OK;
}

API_LIBLOCAL result_e query_delete_iterator_closepage(query_t *query, page_number_t page_number, page_t *page) {
	int j;
	page_item_t i;
	transaction_id_t blocker_current;
	transaction_phase_e phase;
	tuple_t *tuple;


	if (query->delete_ctx.dirty) {
		storage_page_write(query->storage, page_number, page);
	}
	page_destroy(page);
	page = NULL;
	storage_page_rwunlock(query->storage, page_number);
	query->delete_ctx.dirty = 0;

	/*
	 * TODO: Be smarter about this. Instead of writing after each
	 * finished transaction, wait for all of them.
	 */
	while (query->delete_ctx.blockers_c > 0) {
		blocker_current = query->delete_ctx.blockers[0].blocker_id;
		if (transaction_deleteability(query->transaction, query->delete_ctx.blockers[0].blocker_id, &phase) != 0) {
			/* Locking failure, bail out, assume deadlock. */
			free(query->delete_ctx.blockers);
			return RESULT_DEADLOCK;
		}
		
		switch (phase) {
			case COMMITED:
				i = 0;
				while (i < query->delete_ctx.blockers_c) {
					if (query->delete_ctx.blockers[i].blocker_id != blocker_current) {
						i++;
						continue;
					}

					if (i < (query->delete_ctx.blockers_c - 1))
						query->delete_ctx.blockers[i] = query->delete_ctx.blockers[query->delete_ctx.blockers_c - 1];
					query->delete_ctx.blockers_c--;
				}
				break;

			case ABORTED:
				/*
				 * The other transaction aborted, so we have to re-open
				 * the page and ensure that the tuples are deleted by us.
				 */
				storage_page_writelock(query->storage, page_number);
				storage_page_read(query->storage, page_number, &page);

				i = 0;
				while (i < query->delete_ctx.blockers_c) {
					if (query->delete_ctx.blockers[i].blocker_id != blocker_current) {
						i++;
						continue;
					}

					tuple = &page->tuple->tuples[query->delete_ctx.blockers[i].blocker_item];
					tuple->deleted = TUPLE_DELETE;
					tuple->id_delete = query->transaction->id;
					
					for (j = 0; j < query->table->columns_c; j++) {
						if (tuple->atoms[j].type == TUPLE_ATOM_BLOBLINK)
							query_delete_blob(query, *(page_number_t*)tuple->atoms[j].data, query->transaction->id);
					}

					if (i < (query->delete_ctx.blockers_c - 1))
						query->delete_ctx.blockers[i] = query->delete_ctx.blockers[query->delete_ctx.blockers_c - 1];
					query->delete_ctx.blockers_c--;
				}

				storage_page_write(query->storage, page_number, page);
				storage_page_rwunlock(query->storage, page_number);

				page_destroy(page);
				page = NULL;
				break;

			default:
				/*
				 * Something very suspicious happened.
				 */
				return RESULT_EIO;
		}
	}

	free(query->delete_ctx.blockers);
	query->delete_ctx.blockers = NULL;

	storage_page_vacuumunlock(query->storage, page_number);

	return RESULT_OK;
}

API_LIBLOCAL result_e query_delete_execute(query_t *query) {
	result_e result;

	result = query_iterator_prepare(query, query_delete_iterator_openpage, query_delete_iterator_closepage);
	if (result != RESULT_OK)
		return result;

	result = query_filter_prepare(query);
	if (result != RESULT_OK)
		return result;

	while ((result = query_delete_step(query)) == RESULT_OK) {
		/* No-op, the step deals with everything. */
	}

	return result;
}

API_LIBLOCAL result_e query_delete_step(query_t *query) {
	int j;
	result_e result;
	tuple_t *tuple;
	delete_blocker_t blocker;

	while ((result = query_iterator_step(query)) == RESULT_OK) {
		tuple = query->iterator.tuple;

		/* 1. Verify visibility. */
		if (!query_filter_tuple_visible_create(query, tuple))
			continue;
		
		/* 2. Verify predicates. */
		if (!query_filter_tuple_predicates(query))
			continue;

		/* 3. Is it deleted by someone? If no, delete it. If yes, add to blockers. */
		if (tuple->deleted != TUPLE_EXISTING) {
			blocker.blocker_id = tuple->id_delete;
			blocker.blocker_item = query->iterator.pointer.page_item;

			ARRAY_APPEND(query->delete_ctx.blockers, query->delete_ctx.blockers_c, blocker);
		} else {
			tuple->deleted = TUPLE_DELETE;
			tuple->id_delete = query->transaction->id;
			query->delete_ctx.dirty = 1;

			for (j = 0; j < query->table->columns_c; j++) {
				if (tuple->atoms[j].type == TUPLE_ATOM_BLOBLINK)
					query_delete_blob(query, *(page_number_t*)tuple->atoms[j].data, query->transaction->id);
			}
		}

		break;
	}

	return result;
}

API_LIBLOCAL result_e query_delete_finalize(query_t *query) {
	query_filter_finalize(query);
	return query_iterator_finalize(query);
}
